package org.apache.rocketmq.proxy.remoting.activity;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.apache.rocketmq.client.exception.AclException;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.utils.NetworkUtil;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.common.ProxyException;
import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
import org.apache.rocketmq.proxy.common.utils.ExceptionUtils;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.proxy.processor.channel.ChannelProtocolType;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.netty.AttributeKeys;
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;

import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

/**
 * @方法描述：所有活动器的父类，该类实现了NettyRequestProcessor接口，
 * 这意味着该类的对象都可以当作请求处理器注册到Netty服务器中。
 * 在第三版本代码中这个类的大部分功能都都用不到，我就先不添加详细注释了，大家只需要知道Netty服务器处理请求的时候，会先调用该类的processRequest()方法即可
 */
public abstract class AbstractRemotingActivity implements NettyRequestProcessor {

    protected final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
    protected final MessagingProcessor messagingProcessor;
    protected static final String BROKER_NAME_FIELD = "bname";
    protected static final String BROKER_NAME_FIELD_FOR_SEND_MESSAGE_V2 = "n";
    //在创建类的对象时，直接向这个map成员变量中添加了四个异常类码和响应码的键值对数据
    //在调用writeErrResponse()方法回复错误相应时，会根据异常类码从这个map中获取对应的响应码
    private static final Map<ProxyExceptionCode, Integer> PROXY_EXCEPTION_RESPONSE_CODE_MAP = new HashMap<ProxyExceptionCode, Integer>() {
        {
            put(ProxyExceptionCode.FORBIDDEN, ResponseCode.NO_PERMISSION);
            put(ProxyExceptionCode.MESSAGE_PROPERTY_CONFLICT_WITH_TYPE, ResponseCode.MESSAGE_ILLEGAL);
            put(ProxyExceptionCode.INTERNAL_SERVER_ERROR, ResponseCode.SYSTEM_ERROR);
            put(ProxyExceptionCode.TRANSACTION_DATA_NOT_FOUND, ResponseCode.SUCCESS);
        }
    };

    public AbstractRemotingActivity(MessagingProcessor messagingProcessor) {
        this.messagingProcessor = messagingProcessor;
    }

    protected RemotingCommand request(ChannelHandlerContext ctx, RemotingCommand request,
                                      ProxyContext context, long timeoutMillis) throws Exception {
        String brokerName;
        if (request.getCode() == RequestCode.SEND_MESSAGE_V2) {
            if (request.getExtFields().get(BROKER_NAME_FIELD_FOR_SEND_MESSAGE_V2) == null) {
                return RemotingCommand.buildErrorResponse(ResponseCode.VERSION_NOT_SUPPORTED,
                        "Request doesn't have field bname");
            }
            brokerName = request.getExtFields().get(BROKER_NAME_FIELD_FOR_SEND_MESSAGE_V2);
        } else {
            if (request.getExtFields().get(BROKER_NAME_FIELD) == null) {
                return RemotingCommand.buildErrorResponse(ResponseCode.VERSION_NOT_SUPPORTED,
                        "Request doesn't have field bname");
            }
            brokerName = request.getExtFields().get(BROKER_NAME_FIELD);
        }
        if (request.isOnewayRPC()) {
            messagingProcessor.requestOneway(context, brokerName, request, timeoutMillis);
            return null;
        }
        messagingProcessor.request(context, brokerName, request, timeoutMillis)
                .thenAccept(remotingCommand -> writeResponse(ctx, context, request, remotingCommand))
                .exceptionally(t->{
                    writeErrResponse( ctx, context, request, t);
                    return null;
                });
        return null;
    }
    /**
     * @方法描述：Netty构建的服务器处理请求的时候，会先调用该方法。
     */
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request){
        ProxyContext proxyContext = createContext(ctx, request);

        try {
            RemotingCommand response = this.processRequest0(ctx, request, proxyContext);
            if ( response != null){
                writeResponse( ctx, proxyContext, request, response);
            }
            return null;
        } catch (Exception e) {
            writeErrResponse( ctx, proxyContext, request, e);
             return null;
        }
    }

    protected abstract RemotingCommand processRequest0(ChannelHandlerContext ctx, RemotingCommand request, ProxyContext proxyContext) throws Exception;


    public ProxyContext createContext(ChannelHandlerContext ctx, RemotingCommand request) {
        ProxyContext context = ProxyContext.create();
        //得到发送请求的Channel
        Channel channel = ctx.channel();
        //设置请求类型
        context.setAction(RemotingHelper.getRequestCodeDesc(request.getCode()))
                //设置网络传输协议
                .setProtocolType(ChannelProtocolType.REMOTING.getName())
                //设置发送请求过来的channel
                .setChannel(channel)
                //设置本地网络地址，也就是当前proxy模块部署的服务器的地址
                .setLocalAddress(NetworkUtil.socketAddress2String(ctx.channel().localAddress()))
                //设置channel的远程地址
                .setRemoteAddress(RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
        //设置语言类型
        Optional.ofNullable(RemotingHelper.getAttributeValue(AttributeKeys.LANGUAGE_CODE_KEY, channel))
                .ifPresent(language -> context.setLanguage(language.name()));
        //设置客户端Id
        Optional.ofNullable(RemotingHelper.getAttributeValue(AttributeKeys.CLIENT_ID_KEY, channel))
                .ifPresent(context::setClientID);
        //设置客户端版本
        Optional.ofNullable(RemotingHelper.getAttributeValue(AttributeKeys.VERSION_KEY, channel))
                .ifPresent(version -> context.setClientVersion(MQVersion.getVersionDesc(version)));
        return context;

    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/31
     * @方法描述：回复错误响应的方法，在该方法中，会根据异常的类型，设置响应的错误码和错误信息，然后回复出去
     */
    protected void writeErrResponse(ChannelHandlerContext ctx, final ProxyContext context,
                                    final RemotingCommand request, Throwable t) {
        t = ExceptionUtils.getRealException(t);
        if (t instanceof ProxyException) {
            ProxyException e = (ProxyException) t;
            //在这如果判断出当前异常为ProxyException，那么就根据异常的类型，设置响应的错误码和错误信息
            writeResponse(ctx, context, request, RemotingCommand.createResponseCommand(PROXY_EXCEPTION_RESPONSE_CODE_MAP.getOrDefault(e.getCode(), ResponseCode.SYSTEM_ERROR), e.getMessage()), t);
        } else if (t instanceof MQClientException) {
            MQClientException e = (MQClientException) t;
            writeResponse(ctx, context, request, RemotingCommand.createResponseCommand(e.getResponseCode(), e.getErrorMessage()), t);
        } else if (t instanceof MQBrokerException) {
            MQBrokerException e = (MQBrokerException) t;
            writeResponse(ctx, context, request, RemotingCommand.createResponseCommand(e.getResponseCode(), e.getErrorMessage()), t);
        } else if (t instanceof AclException) {
            writeResponse(ctx, context, request, RemotingCommand.createResponseCommand(ResponseCode.NO_PERMISSION, t.getMessage()), t);
        } else {
            writeResponse(ctx, context, request,
                    RemotingCommand.createResponseCommand(ResponseCode.SYSTEM_ERROR, t.getMessage()), t);
        }
    }

    protected void writeResponse(ChannelHandlerContext ctx, final ProxyContext context,
                                 final RemotingCommand request, RemotingCommand response) {
        writeResponse(ctx, context, request, response, null);
    }


    /**
     * @方法描述：回复响应的方法
     */
    private void writeResponse(ChannelHandlerContext ctx, ProxyContext context,
                               RemotingCommand request, RemotingCommand response, Throwable t) {
        //判断请求类型是否为单项请求，如果是单项请求，说明客户端不关注响应，那么直接退出当前方法即可，不必回复响应
        if (request.isOnewayRPC()) {
            return;
        }
        //判断当前channel是否可写，不可写则直接退出当前方法
        if (!ctx.channel().isWritable()) {
            return;
        }
        ProxyConfig config = ConfigurationManager.getProxyConfig();
        //把请求唯一Id设置到响应中
        response.setOpaque(request.getOpaque());
        //设置RemotingCommand消息对象为响应对象
        response.markResponseType();
        //设置消息所属的区域Id
        response.addExtField(MessageConst.PROPERTY_MSG_REGION, config.getRegionId());
        //设置消息追踪开关，默认是不开启的
        response.addExtField(MessageConst.PROPERTY_TRACE_SWITCH, String.valueOf(config.isTraceOn()));
        //如果有异常，那就把异常信息设置到响应的备注信息中
        if (t != null) {
            response.setRemark(t.getMessage());
        }
        //回复响应消息
        ctx.writeAndFlush(response);
    }
}
